#include "Xps/Protocol/TCP/Server.h"

namespace Xps
{
    namespace TCP
    {
        Server::Server(const std::string &_Name, const short &_Port, const IObjectFactory *_ObjectFactory )
            :name(_Name), port(_Port), is_running(false), object_factory(_ObjectFactory)
        {
            Initialize();
        }

        void Server::Start()
        {
            if(!is_running)
            {
                is_running = true;
                io_service = boost::shared_ptr<asio::io_service>(new asio::io_service());
                acceptor = boost::shared_ptr<tcp::acceptor>(new tcp::acceptor(*io_service, tcp::endpoint(tcp::v4(), port)));
                th_listner = boost::thread(&Server::Listner,this);
                th_sender = boost::thread(&Server::Sender,this);
                th_sweeper = boost::thread(&Server::Sweeper,this);
                th_process_packets = boost::thread(&Server::ProcessPackets, this);
            }
        }

        void Server::Stop()
        {
            if(is_running)
            {
                is_running = false;
                io_service->stop();
                SecureIn();
                for(ClientSessionIterator i = connected_clients.begin();
                    i != connected_clients.end(); i++)
                {
                    ClientSession::Ptr client_session = *i;
                    client_session->Stop();
                }
                SecureOut();
                TerminateThread(th_listner.native_handle(),0);
                TerminateThread(th_sender.native_handle(),0);
                TerminateThread(th_sweeper.native_handle(),0);
                TerminateThread(th_process_packets.native_handle(),0);
            }
        }

        void Server::Send( IPacket::Ptr _Packet )
        {
            outgoing_queue.push(_Packet);
            SetEvent(sender_event);

        }

        void Server::Listner()
        {
            while(is_running)
            {
                ClientSession::Ptr client_session(new ClientSession(*io_service, incomming_queue, recieve_event, object_factory));
                acceptor->accept(client_session->getSocket());
                SecureIn();
                {
                    tcp::endpoint remote_endpoint = client_session->getSocket().remote_endpoint();
                    connected_clients.push_back(client_session);
                }
                SecureOut();
                client_session->Start();
            }
        }

        void Server::Sender()
        {
            while(is_running)
            {
                WaitForSingleObject(sender_event, INFINITE);
                SecureIn();
                {
                    for(int i=0,
                        size = outgoing_queue.size();
                        i<size;i++)
                    {
                        IPacket::Ptr packet = outgoing_queue.pop();

                        if(connected_clients.size()>1)
                        {
                            int k=0;
                        }
                        for(ClientSessionIterator i = connected_clients.begin();
                            i != connected_clients.end(); i++)
                        {
                            ClientSession::Ptr client_session = *i;
                            client_session->Send(packet);
                        }
                    }
                }
                SecureOut();
            }
        }

        void Server::Sweeper()
        {
            while(is_running)
            {        
                SecureIn();
                {
                    static concurrent_queue<ClientSessionIterator > to_remove;
                    for(ClientSessionIterator i = connected_clients.begin();
                        i != connected_clients.end(); i++)
                    {
                        ClientSession::Ptr client_session = *i;
                        if(!client_session->IsRunning())
                            to_remove.push(i);
                    }
                    while(!to_remove.empty())
                    {
                        ClientSessionIterator i = to_remove.pop();
                        connected_clients.erase(i);
                    }
                }
                SecureOut();
                Sleep(200);
            }
        }

        void Server::ProcessPackets()
        {
            while(is_running)
            {
                WaitForSingleObject(recieve_event,INFINITE);
                for(int i=0,
                    size = incomming_queue.size();
                    i<size;i++)
                {
                    const IObject::Ptr object = incomming_queue.pop();
                    __raise OnPacketRecieved(name, object.get());

                }
            }
        }

        const std::string & Server::getName() const
        {
            return name;
        }

        Server::~Server( void )
        {
            Stop();
            // Release resources used by the critical section object.
            DeleteCriticalSection(&critical_section);
            CloseHandle(sender_event);
            CloseHandle(recieve_event);
        }

        void Server::Initialize()
        {
            // Initialize the critical section one time only.
            if (!InitializeCriticalSectionAndSpinCount(&critical_section, 
                0x00004000) ) 
                throw std::exception("concurrent queue initialize error!");

            sender_event = CreateEvent(NULL, FALSE, FALSE, NULL);
            recieve_event = CreateEvent(NULL, FALSE, FALSE, NULL);
        }

        void Server::SecureIn()
        {
            EnterCriticalSection(&critical_section);
        }

        void Server::SecureOut()
        {
            LeaveCriticalSection(&critical_section);
        }
    };
};